package com.funcourse.demo.api.websocket;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.funcourse.demo.api.restful.io.response.busi.CompeteRoomSocketVO;
import com.funcourse.demo.api.websocket.SocketMessage.SocketMessageType;
import com.funcourse.demo.business.RoomCacher;
import com.funcourse.demo.model.CompeteRoom;
import com.funcourse.demo.model.CompeteRoom.RoomStatus;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

@Slf4j
@Component
public class SocketServer extends TextWebSocketHandler {

  Map<String, Set<String>> roomIdSessionsMap = new HashMap<>();

  Map<String, WebSocketSession> socketSessionHolder = new HashMap<>();

  ObjectMapper objectMapper = new ObjectMapper();

  @Autowired
  RoomCacher roomCacher;

  private static final String HEART_BEAT_MSG="heart beat";

  @Override
  public void afterConnectionEstablished(WebSocketSession session) throws Exception {
    socketSessionHolder.put(session.getId(), session);
    final CompeteRoom competeRoom = (CompeteRoom)session.getAttributes().get(WebSocketHandShakeInterceptor.SESSION_ROOM);
    Set<String> webSocketSessions = roomIdSessionsMap.get(competeRoom.getId());
    if (null == webSocketSessions) {
      webSocketSessions = new LinkedHashSet<>();
      roomIdSessionsMap.put(competeRoom.getId(), webSocketSessions);
    }
    webSocketSessions.add(session.getId());

    log.info(" room connection established {} {}",competeRoom.getId(),session.getId());
    if(RoomStatus.CLOSED == competeRoom.getRoomStatus()){
      try {
        sendTextMsgPayload(objectMapper.writeValueAsString(new SocketMessage(
            SocketMessageType.ROOM_CLOSE, competeRoom.getId())),session);
        session.close();
      } catch (JsonProcessingException e) {
        log.error("heart beat send error",e);
      } catch (IOException e) {
        log.error("close web socket session error",e);
      }
      return;
    }
    this.sendRoomUpdateMessage(competeRoom,session);
  }

  @Override
  public void handleTextMessage(WebSocketSession session, TextMessage message) {
    String messageReceived = message.getPayload();
    if(HEART_BEAT_MSG.equalsIgnoreCase(messageReceived)){
      log.debug("heart beat received from {}",session.getId());
      try {
      sendTextMsgPayload(objectMapper.writeValueAsString(new SocketMessage(
          SocketMessageType.HEART_BEAT, HEART_BEAT_MSG + " "+ session.getId() + " "+ System.currentTimeMillis())),session);
      } catch (IOException e) {
        log.error("heart beat send error");
      }
      return;
    }
  }

  @Override
  public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
    log.debug("socket closed {} {}",session.getId(),session.getAttributes().get(WebSocketHandShakeInterceptor.SESSION_USER_ID));
    socketSessionHolder.remove(session.getId());
  }

  public List<WebSocketSession> getSessionsOfRoom(String roomId) {
    Set<String> sessionIds = roomIdSessionsMap.get(roomId);
    if (CollectionUtils.isEmpty(sessionIds)) {
      return new ArrayList<>();
    }
    sessionIds.removeIf(sid->{
      final boolean sessionIsNull = null == socketSessionHolder.get(sid);
      if(sessionIsNull){
        log.debug("remove session {} from room {} because it is null",sid,roomId);
      }
      return sessionIsNull;
    });
    log.debug("after remove null sessions for room {} now have session count {}",roomId,sessionIds.size());
    return sessionIds.stream().map(sid -> {
      WebSocketSession webSocketSession = socketSessionHolder.get(sid);
      return webSocketSession;
    }).filter(s->null != s).collect(Collectors.toList());
  }

  public void sendRoomStatusChangeMsg(String competeRoomId, RoomStatus roomStatus){
    if(null == competeRoomId || null == roomStatus){
      log.error("bad room status {} or bad competeRoomId",roomStatus,competeRoomId);
      return;
    }
    List<WebSocketSession> sessionsOfRoom = getSessionsOfRoom(competeRoomId);
    log.debug("total connections for sendRoomStatusChangeMsg roomId {} {}",competeRoomId,sessionsOfRoom.size());
    SocketMessageType messageType;
    if(RoomStatus.STARTED == roomStatus){
      messageType = SocketMessageType.ROOM_START;
    }else if (RoomStatus.CLOSED == roomStatus){
      messageType = SocketMessageType.ROOM_CLOSE;
    }else{
      log.error("bad room status {}",roomStatus);
      return;
    }
    String payload;
    try {
      payload = objectMapper.writeValueAsString(new SocketMessage(
          messageType, competeRoomId));
    } catch (JsonProcessingException e) {
      log.error("msg serilization failed {}","room started " + competeRoomId,e);
      return;
    }
    sessionsOfRoom.forEach(s -> {
      sendTextMsgPayload(payload, s);
    });
  }

  public void sendTestQuestionMsg(String competeRoomId,Integer questionId,SocketMessageType msgType){
    List<WebSocketSession> sessionsOfRoom = getSessionsOfRoom(competeRoomId);
    log.debug("total connections for sendTestQuestionMsg roomId {} {}",competeRoomId,sessionsOfRoom.size());
    Map<String,Object> msgContent=new HashMap<>();
    msgContent.put("competeRoomId",competeRoomId);
    msgContent.put("questionId", questionId);
    String payload;
    try {
      payload = objectMapper.writeValueAsString(new SocketMessage(
          msgType, msgContent));
    } catch (JsonProcessingException e) {
      log.error("msg serilization failed {}",msgContent);
      return;
    }
    sessionsOfRoom.forEach(s->{
      sendTextMsgPayload(payload,s);
    });
  }

  public void sendRoomUpdateMessage(String roomId) {
    CompeteRoom competeRoom = roomCacher.getRoomById(roomId);
    if(null != competeRoom){
      sendRoomUpdateMessage(competeRoom,null);
    }

  }

  private void sendRoomUpdateMessage(CompeteRoom competeRoom,WebSocketSession session){
    List<WebSocketSession> sessionsOfRoom =
        session == null ? getSessionsOfRoom(competeRoom.getId()) : Arrays.asList(session);
    if(null != sessionsOfRoom){
      log.debug("total connections for roomId {} {}",competeRoom.getId(),sessionsOfRoom.size());
    }else{
      log.debug("no connections found for room {}",competeRoom.getId());
      return;
    }

    sessionsOfRoom.forEach(s->{
      final String userId = (String)s.getAttributes().get(WebSocketHandShakeInterceptor.SESSION_USER_ID);
      final Integer topUserAmount = (Integer)s.getAttributes().get(WebSocketHandShakeInterceptor.SESSION_TOP_USER_AMOUNT);
      String payload;
      try {
        payload = objectMapper.writeValueAsString(new SocketMessage(
            SocketMessageType.ROOM_UPDATE, new CompeteRoomSocketVO(competeRoom,userId,topUserAmount,false)));
      } catch (JsonProcessingException e) {
        log.error("msg serilization failed {}",competeRoom);
        return;
      }
      final long timeMillis = System.currentTimeMillis();
      sendTextMsgPayload(payload, s);
      final long timeMillisEnd = System.currentTimeMillis();
      log.debug("socket send for room caused {} {}",competeRoom.getId(),timeMillisEnd-timeMillis);
    });
  }

  private void sendTextMsgPayload(String payload, WebSocketSession s) {
    try {
      TextMessage textMessage = new TextMessage(payload);
      synchronized (s){
        s.sendMessage(textMessage);
      }
    } catch (IOException e) {
      log.error("sending room start command to {} failed", s.getId(), e);
      socketSessionHolder.remove(s.getId());
    }
  }
}
